package elephant.rpc.network.netty;



import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import elephant.rpc.network.netty.codec.FastJsonMessageDecoder;
import elephant.rpc.network.netty.codec.FastJsonMessageEncoder;
import elephant.rpc.network.netty.codec.JDKSerialMessageDecoder;
import elephant.rpc.network.netty.codec.JDKSerialMessageEncoder;
import elephant.rpc.server.RPCServer;
import elephant.rpc.server.core.NetworkService;
import elephant.rpc.threadpool.RPCThreadFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.compression.ZlibCodecFactory;
import io.netty.handler.codec.compression.ZlibWrapper;
import io.netty.handler.codec.serialization.ClassResolvers;

/**
 * 
 * @author skydu
 *
 */
public class NettyServerService extends NetworkService{
	//
	private static Logger logger=LoggerFactory.getLogger(NettyServerService.class);
	//
	private ServerBootstrap serverStrap;
	private EventLoopGroup bossGroup;
	private EventLoopGroup workerGroup;
	private NettyServerHandler serverHandler;
	//
	public NettyServerService(RPCServer rpcServer) {
		super(rpcServer);
		serverHandler=new NettyServerHandler(this);
	}

	@Override
	public void init() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" init");
		}
	}

	@Override
	public void start() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" start");
		}
		int cpuNum=Runtime.getRuntime().availableProcessors();
		bossGroup = new NioEventLoopGroup(0,Executors.newFixedThreadPool(
				cpuNum+1,new RPCThreadFactory("RPCBossNioWorker",
						rpcServer.getContextClassLoader())));
	    workerGroup = new NioEventLoopGroup(0,Executors.newFixedThreadPool(
	    		cpuNum+1,new RPCThreadFactory("RPCWorkNioWorker",
	    				rpcServer.getContextClassLoader())));
	    //
	    serverStrap = new ServerBootstrap();
	    serverStrap.group(bossGroup,workerGroup)
         .channel(NioServerSocketChannel.class)
         .option(ChannelOption.SO_BACKLOG, 100)
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) throws Exception {
            	 ChannelPipeline pipeline=ch.pipeline();
            	 //Enable stream compression (you can remove these two if unnecessary)
            	 pipeline.addLast(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));
            	 pipeline.addLast(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
            	 //
            	 String codecType=rpcServer.getCodecType();
            	 if(RPCServer.CODEC_TYPE_JDK.equals(codecType)){
 					if(logger.isDebugEnabled()){
 						logger.debug("codec type jdk");
 					}
 					pipeline.addLast(
 						new JDKSerialMessageEncoder(),
 						new JDKSerialMessageDecoder(ClassResolvers.softCachingConcurrentResolver(
 								Thread.currentThread().getContextClassLoader())),
 								serverHandler);
 				 }
            	 if(RPCServer.CODEC_TYPE_FASTJSON.equals(codecType)){
            		 if(logger.isDebugEnabled()){
  						logger.debug("codec type fastjson");
  					}
            		 pipeline.addLast(
            			new FastJsonMessageEncoder(performMonitor),
                    	new FastJsonMessageDecoder(performMonitor),
                             	serverHandler);
            	 }
                
             }
         });
		serverStrap.bind(rpcServer.getPort()).sync();
	}

	@Override
	public void stop() throws Exception {
		if(logger.isInfoEnabled()){
			logger.info(getClass().getSimpleName()+" stop");
		}
		if(bossGroup!=null){
			try {
				bossGroup.shutdownGracefully();
			} catch (Exception e) {
				logger.error(e.getMessage(),e);
			}
		}
		if(workerGroup!=null){
			try {
				workerGroup.shutdownGracefully();
			} catch (Exception e) {
				logger.error(e.getMessage(),e);
			}
		}
	}

}
